﻿using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.ObjectPool;
using RabbitMQ.Client;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Text;

namespace QQZiFramework.MQ
{
    public class RabbitMQPoolChannel : MQIModel
    {
        public RabbitMQConnectShareManage.ConnectChannelInfo Connection { get; set; }
        public ObjectPool<RabbitMQPoolChannel> Pool { get; internal set; }
        public DateTime? ExpirationTime { get; internal set; }

        public DateTime CreateTime;


        public override void Dispose()
        {
            RabbitMQChannelPoolManager.ReturnChannel(this);
        }

        // public void Close()
        // {
        //     RabbitMQChannelPoolManager.ReturnChannel(this);
        // }
    }
    internal class RabbitMQChannelPoolManager
    {
        internal static ushort BasicQos;
        static int maxPoolSize;
        static IConfiguration configuration;
        static int expirationSeconds;

        private static ConcurrentDictionary<string, ObjectPool<RabbitMQPoolChannel>> _channelPools = new ConcurrentDictionary<string, ObjectPool<RabbitMQPoolChannel>>();

        public static RabbitMQPoolChannel GetChannel(string queueName)
        {
            ObjectPool<RabbitMQPoolChannel> pool;
            lock (_channelPools)
            {
                if (configuration == null)
                {
                    configuration = DBContext.ServiceProvider.GetService(typeof(IConfiguration)) as IConfiguration;
                    var configItem = configuration.GetSection("RabbitMQ");
                    if (configItem["ChannelMaxPoolSize"] != null) maxPoolSize = Convert.ToInt32(configItem["ChannelPoolSize"]);
                    else maxPoolSize = 500;
                    if(configItem["ChannelPoolExpirationSeconds"] !=null) expirationSeconds = Convert.ToInt32(configItem["ChannelPoolExpirationSeconds"]);
                    else expirationSeconds = 60;
                }
                if (!_channelPools.TryGetValue(queueName, out pool))
                {
                    var channelPoolPolicy = new RabbitMQPooledObjectPolicy(queueName, configuration);
                    pool = new DefaultObjectPool<RabbitMQPoolChannel>(channelPoolPolicy, maxPoolSize);
                    _channelPools[queueName] = pool;
                }
            }
            var channel = pool.Get();
            while (channel.ExpirationTime != null && channel.ExpirationTime.Value < DateTime.Now)
            {
                //空闲超时的
                pool.Return(channel);
                channel = pool.Get();
            }
            channel.Pool = pool;
            channel.ExpirationTime = DateTime.Now.AddSeconds(expirationSeconds);//超过该时间，不复用
            return channel;
        }

        /// <summary>
        /// 放回连接池
        /// </summary>
        /// <param name="channel"></param>
        public static void ReturnChannel(RabbitMQPoolChannel channel)
        {
            channel.Pool.Return(channel);
        }
    }

    public class RabbitMQPooledObjectPolicy : DefaultPooledObjectPolicy<RabbitMQPoolChannel>
    {
        public string queueName;
        bool Durable = true;//默认不持久化
        DBContext context;
        public RabbitMQPooledObjectPolicy(string queueName, IConfiguration configuration)
        {
            this.queueName = queueName;
            this.configuration = configuration;
            context = new DBContext();
        }
        IConfiguration configuration;


        private RabbitMQPoolChannel CreateMQIModel()
        {
            context.LogDebug("创建Channel:" + this.queueName, "RabbitMQ");
            var section = configuration.GetSection("RabbitMQ:Queues:" + queueName);
            // var conn = connectionFactory.CreateConnection();    
            var conn = RabbitMQConnectShareManage.GetConnection();
            var channel = conn.CreateChannel();//创建连接会话对象
            bool durable = (section != null && !string.IsNullOrEmpty(section["Durable"])) ? Convert.ToBoolean(section["Durable"]) : Durable;
            var bindSection = section.GetSection("Bind");
            string exchange = null;
            string routingKey = null;
            if (bindSection != null)
            {
                exchange = bindSection["exchange"];
                if (exchange == null) exchange = "";
            }

            ushort basicQos = (section != null && !string.IsNullOrEmpty(section["BasicQos"])) ? Convert.ToUInt16(section["BasicQos"]) : RabbitMQChannelPoolManager.BasicQos;
            if (basicQos > 0) channel.BasicQos(0, basicQos, false);
            IBasicProperties properties = null;
            if (durable)
            {
                //消息持久化
                properties = channel.CreateBasicProperties();
                properties.DeliveryMode = 2;
            }
            if (exchange != null && exchange.Length > 0)
            {
                //交换机绑定，如果配置名与交换机名一致，这发交换机，如果不一致，发队列
                routingKey = bindSection["routingKey"];
                if (routingKey == null) routingKey = "#";
                string type = bindSection["exchangeType"];
                if (type == null) type = "topic";
                //申明交换机
                channel.ExchangeDeclare(exchange, type, durable);
                if (exchange != queueName)
                {
                    //发布到交换机的某个队列
                    //声明一个队列            
                    channel.QueueDeclare(
                        queue: queueName,
                         durable: durable,//是否持久化,true持久化,队列会保存磁盘,服务器重启时可以保证不丢失相关信息。
                         exclusive: false,//是否排他,true排他的,如果一个队列声明为排他队列,该队列仅对首次声明它的连接可见,并在连接断开时自动删除.
                         autoDelete: false,//是否自动删除。true是自动删除。自动删除的前提是：致少有一个消费者连接到这个队列，之后所有与这个队列连接的消费者都断开时,才会自动删除.
                         arguments: null //设置队列的一些其它参数
                      );
                    //队列与交换机绑定
                    channel.QueueBind(queueName, exchange, routingKey);
                    exchange="";
                    routingKey=queueName;
                }
            }
            else
            {
                //声明一个队列
                channel.QueueDeclare(
                    queue: queueName,
                     durable: durable,//是否持久化,true持久化,队列会保存磁盘,服务器重启时可以保证不丢失相关信息。
                     exclusive: false,//是否排他,true排他的,如果一个队列声明为排他队列,该队列仅对首次声明它的连接可见,并在连接断开时自动删除.
                     autoDelete: false,//是否自动删除。true是自动删除。自动删除的前提是：致少有一个消费者连接到这个队列，之后所有与这个队列连接的消费者都断开时,才会自动删除.
                     arguments: null //设置队列的一些其它参数
                  );
                  routingKey=queueName;
            }
            return new RabbitMQPoolChannel { Connection = conn, Channel = channel, DurableProperties = properties, Exchange = exchange, RoutingKey = routingKey };
        }

        public override RabbitMQPoolChannel Create()
        {
            var channel = CreateMQIModel();
            channel.CreateTime = DateTime.Now;
            return channel;
        }

        /// <summary>
        /// 放回池子
        /// </summary>
        /// <param name="obj"></param>
        /// <returns>是否可复用</returns>
        public override bool Return(RabbitMQPoolChannel obj)
        {
            bool ret = obj.ExpirationTime > DateTime.Now;
            if (!ret)
            {
                obj.Connection.ReleaseChannel(obj.Channel);
                context.LogDebug("Channel超过时间，从Channel连接池移除", "RabbitMQ");
            }
            return ret;
        }

    }

}
